/*
 * 文件名：RealStreamCTLHandler.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：yuanze
 * 修改时间：2014年10月31日
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.protocal.aipr.netty.handler;

import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.realstream.server.model.RealStreamUser;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;

/**
 * @description 实时流命令端口心跳包处理
 * @author [yuanze]
 * @version [版本号,2014年10月31日]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
@Sharable
public class RealStreamHeartHandler extends SimpleChannelInboundHandler<String>
{
	private static final Logger logger = LoggerFactory.getLogger(RealStreamHeartHandler.class);

	/**
	 * @see io.netty.channel.ChannelInboundHandlerAdapter#channelInactive(io.netty.channel.ChannelHandlerContext)
	 */
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception
	{
		String address = ctx.channel().remoteAddress().toString();
		RealStreamUser user = RealStreamTool.AUTHENTICATION_INFO_CACHE.remove(address);
		if (user != null)
		{
			RealStreamTool.AUTHORISED_USER_INFO.remove(user.getClientIP() + ":" + user.getDataConnPort());
			ChannelHandlerContext channelHandler = RealStreamTool.ACCEPTING_CHANNEL_CACHE.remove(user.getClientIP() + ":" + user.getDataConnPort());
			if (channelHandler != null)
			{
				channelHandler.channel().close();
			}
		}
		logger.info("监测到remote[" + ctx.channel().remoteAddress() + "]关闭,处理缓存信息");
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception
	{
		String address = ctx.channel().remoteAddress().toString();
		if (!RealStreamTool.USER_HEART_COUNT.contains(address))
		{
			RealStreamTool.USER_HEART_COUNT.put(address, 3);
		}
		// 过滤心跳包
		if (StringUtils.isEmpty(msg) || msg.startsWith("HEART"))
		{
			logger.debug("接受到客户端回复心跳包信息[" + msg + "]");
		}
		else
		{
			ctx.fireChannelRead(msg);
		}
	}

	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception
	{
		ctx.flush();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
	{
		String address = ctx.channel().remoteAddress().toString();
		RealStreamUser user = RealStreamTool.AUTHENTICATION_INFO_CACHE.remove(address);
		if (user != null)
		{
			RealStreamTool.AUTHORISED_USER_INFO.remove(user.getClientIP() + ":" + user.getDataConnPort());
			ChannelHandlerContext channelHandler = RealStreamTool.ACCEPTING_CHANNEL_CACHE.remove(user.getClientIP() + ":" + user.getDataConnPort());
			if (channelHandler != null)
			{
				channelHandler.channel().close();
			}
		}
		RealStreamTool.USER_HEART_COUNT.remove(address);
		logger.error("Unexpected exception from downstream.", cause);
		logger.info("监测到remote[" + ctx.channel().remoteAddress() + "]连接异常关闭,处理缓存信息");
		ctx.close();
	}

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception
	{
		// Channel channel = ctx.channel();
		// String address = ctx.channel().remoteAddress().toString();
		// if (evt instanceof IdleStateEvent)
		// {
		// IdleStateEvent e = (IdleStateEvent) evt;
		// if (e.state() == IdleState.ALL_IDLE)
		// {
		// logger.info("检测到客户端[" + address + "]channel空闲发送心跳包监测");
		// int count = RealStreamTool.USER_HEART_COUNT.get(address);
		// logger.info("连续发送客户端[" + channel.remoteAddress() + "]心跳包到客户端未回复" +
		// count + "次");
		// // 心跳次数达到一定时间主动断开连接
		// if (++count >= 3)
		// {
		// logger.warn("连续发送心跳包到客户端3次未响应，断开数据连接......");
		// // 通知关闭数据端口通道
		// RealStreamUser user =
		// RealStreamTool.AUTHENTICATION_INFO_CACHE.remove(address);
		// if (user != null)
		// {
		// RealStreamTool.AUTHORISED_USER_INFO.remove(user.getClientIP() + ":" +
		// user.getDataConnPort());
		// ChannelHandlerContext channelHandler =
		// RealStreamTool.ACCEPTING_CHANNEL_CACHE.remove(user.getClientIP() +
		// ":"
		// + user.getDataConnPort());
		// if (channelHandler != null)
		// {
		// channelHandler.channel().close();
		// }
		// }
		// RealStreamTool.USER_HEART_COUNT.remove(address);
		// ctx.close();
		// }
		// else
		// {
		// ctx.writeAndFlush("HEART" + Constant.LINE_SEPARATOR_UNIX);
		// }
		// }
		// }
	}
}
